package com.etlmaster.hbase.rowProcess;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.FilterList.Operator;
import org.apache.hadoop.hbase.util.Bytes;

import com.etlmaster.executor.bigdata.hbase.HbaseUtil;
import com.etlmaster.executor.trans.core.Line;
import com.etlmaster.executor.utils.LogWriter;
import com.etlmaster.rest.server_rowProcess.ServerRests_RowProcess;
import com.ibm.icu.util.Calendar;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
public class HbaseUtil_rowProcess extends HbaseUtil {

	public static void put(HTable table, Line line, String rowKey) {

		LogWriter.addLog("INFO","更新表:{},key:{}",table.getName().getNameAsString(),rowKey);

		List<Put> puts=new ArrayList<Put>();
		Put put = new Put(Bytes.toBytes(rowKey));
		
		for(int i=0;i<line.getMeta().getCounter();i++){
			String columnName =line.getMeta().getFieldMetas()[i].getFieldName().toLowerCase();
			String columnValue = line.getField(i);
			put.add(Bytes.toBytes(columnName), Bytes.toBytes(columnName),
					Bytes.toBytes(columnValue));
		}
		puts.add(put);
		putList(table, puts);

	}

	public  static String getCurrentRowKey(HTable table, String columnFamily,
			String label, String starter, String value) {
		String result = "";
		try {
			// HTable table = new HTable(getConf(), tableName);
			Scan s = new Scan();
			RowFilter sf1 = new RowFilter(CompareOp.EQUAL,
					new RegexStringComparator(starter));
			SingleColumnValueFilter sf2 = new SingleColumnValueFilter(
					Bytes.toBytes(columnFamily), Bytes.toBytes(label),
					CompareOp.EQUAL, Bytes.toBytes(value));

			FilterList filter = new FilterList(Operator.MUST_PASS_ALL, sf1, sf2);
			s.setFilter(filter);
			ResultScanner rs = table.getScanner(s);
			for (Result r : rs) {
				result = new String(r.getRow());
			}
			rs.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return result;
	}
	public  static String getCurrentRowKey(HTable table, String starter) {
		String result = "";
		try {
			// HTable table = new HTable(getConf(), tableName);
			Scan s = new Scan();
			RowFilter sf1 = new RowFilter(CompareOp.EQUAL,
					new RegexStringComparator(starter));
			
			FilterList filter = new FilterList(Operator.MUST_PASS_ALL, sf1);
			s.setFilter(filter);
			ResultScanner rs = table.getScanner(s);
			for (Result r : rs) {
				result = new String(r.getRow());
			}
			rs.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return result;
	}
	public synchronized static String getMaxRowKey(HTable table, String starter) {
		String result = "";
		if (ServerRests_RowProcess.getMaxRowKey().get(starter) == null) {
			try {
				Scan s = new Scan();
				RowFilter sf1 = new RowFilter(CompareOp.EQUAL,
						new RegexStringComparator(starter));

				FilterList filter = new FilterList(Operator.MUST_PASS_ALL, sf1);
				s.setFilter(filter);
				s.setReversed(true);
				ResultScanner rs = table.getScanner(s);
				for (Result r : rs) {
					result = new String(r.getRow());
					break;
				}
				rs.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
			if (result.equalsIgnoreCase("")) {
				result = starter + " 00:00:00";
			} else {
				result = addOneSec(result);
			}
		} else {
			result = addOneSec(ServerRests_RowProcess.getMaxRowKey().get(starter));
		}
		ServerRests_RowProcess.getMaxRowKey().put(starter, result);

		return result;
	}

	public static String addOneSec(String result) {
		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		try {
			Date date = sdf.parse(result);
			Calendar ca = Calendar.getInstance();
			ca.setTime(date);
			ca.add(Calendar.SECOND, 1);
			Date new_date = ca.getTime();
			return sdf.format(new_date);

		} catch (ParseException e) {
			e.printStackTrace();
		}
		return null;
	}

}
